package xin.marcher.module.seller.mq.consumer.listener;

import com.alibaba.nacos.common.utils.CollectionUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;
import xin.marcher.module.common.constants.seller.SellerRedisKeyConstants;
import xin.marcher.module.common.enums.seller.SellerTypeEnum;
import xin.marcher.module.common.exception.BaseBizException;
import xin.marcher.module.common.exception.ProductErrorCodeEnum;
import xin.marcher.module.common.redis.RedisCache;
import xin.marcher.module.common.utils.JsonUtil;
import xin.marcher.module.seller.domain.response.SellerInfoResponse;
import xin.marcher.module.seller.mq.message.PreheatSellerMessage;

import javax.annotation.Resource;
import java.util.*;

/**
 * 缓存预热消费者
 */
@Slf4j
@Component
public class SellerPreheatCacheListener implements MessageListenerConcurrently {

    @Resource
    private RedisCache redisCache;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        // 每个消费者每次消费一页数据，也就是一条消息，均量消费，这条消息是List<String>，每个string包含页码和该页下的SellerInfo集合
        // 拿到消息之后
        // 遍历消息
        // 构造出多个SellerInfo的Map，key是redis的key，value是对象
        // 转换出sellerIdList的集合，set到卖家类型缓存
        try {
            for (MessageExt messageExt : list) {
                String msg = new String(messageExt.getBody());
                List messageList = JsonUtil.json2Object(msg, List.class);
                if (CollectionUtils.isEmpty(messageList)) {
                    throw new BaseBizException(ProductErrorCodeEnum.PARAM_CHECK_ERROR, ProductErrorCodeEnum.PARAM_CHECK_ERROR.getErrorCode());
                }

                Map<String, String> sellerInfoResultMap = new HashMap<>(messageList.size());
                List<Long> sellerIdList = new ArrayList<>(messageList.size());
                // 每一个List中的卖家类型都是一样的，哪怕发送的是全部类型的卖家消息，消费者这里，拿到的也还是同一类型的卖家集合
                for (Object message : messageList) {
                    PreheatSellerMessage preheatSellerMessage = JsonUtil.json2Object(message.toString(), PreheatSellerMessage.class);
                    if (Objects.isNull(preheatSellerMessage)) {
                        throw new BaseBizException(ProductErrorCodeEnum.PARAM_CHECK_ERROR, ProductErrorCodeEnum.PARAM_CHECK_ERROR.getErrorCode());
                    }

                    // 卖家信息
                    SellerInfoResponse sellerInfo = preheatSellerMessage.getSellerInfo();
                    // 卖家信息的Key
                    String sellerInfoKey = SellerRedisKeyConstants.SELLER_INFO_LIST + sellerInfo.getSellerId();
                    // 卖家信息集合
                    sellerInfoResultMap.put(sellerInfoKey, JsonUtil.object2Json(sellerInfo));
                    // 卖家ID集合
                    sellerIdList.add(sellerInfo.getSellerId());
                }

                PreheatSellerMessage preheatSellerMessage = JsonUtil.json2Object(messageList.get(0).toString(), PreheatSellerMessage.class);
                if (Objects.isNull(preheatSellerMessage)) {
                    throw new BaseBizException(ProductErrorCodeEnum.PARAM_CHECK_ERROR, ProductErrorCodeEnum.PARAM_CHECK_ERROR.getErrorCode());
                }

                // 卖家类型
                Integer sellerType = preheatSellerMessage.getSellerInfo().getSellerType();
                // 卖家类型的key
                String sellerTypeKey = "";
                if (sellerType.equals(SellerTypeEnum.SELF.getCode())) {
                    sellerTypeKey = SellerRedisKeyConstants.SELF_TYPE_LIST + preheatSellerMessage.getCachePageNo();
                } else {
                    sellerTypeKey = SellerRedisKeyConstants.POP_TYPE_LIST + preheatSellerMessage.getCachePageNo();
                }

                // 将卖家信息set到缓存
                redisCache.mset(sellerInfoResultMap);
                // 将卖家类型ID集合set到缓存
                redisCache.set(sellerTypeKey, JsonUtil.object2Json(sellerIdList), -1);
            }
        } catch (Exception e) {
            log.error("consume error, 缓存预热消息消费失败", e);
            // 本次消费失败，下次重新消费
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
